package com.lee.commonerror.code.concurrent;

import com.lee.commonerror.common.annotation.Right;
import com.lee.commonerror.common.annotation.Wrong;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.Assert;
import org.springframework.util.StopWatch;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.LongStream;

/**
 * 常见错误:
 * 1. 方法分开执行,方法本身原子性,但是好多个方法连续使用整体就没有原子性了
 * 2. 没有使用组合方法,性能大大降低
 * @author ：liylw
 * @date ：Created in 2021/4/9 18:31
 */
@RestController
@RequestMapping("ConcurrentHashMapDemo")
@Slf4j
public class ConcurrentHashMapDemo {

    private static final int THREAD_COUNT = 10;

    private static final int ITEM_COUNT = 1000;

    //循环次数
    private static final int LOOP_COUNT = 10_000_000;

    /**
     * 用来获得一个指定元素数量模拟数据的concurrentHashMap
     * @param count
     * @return
     */
    private ConcurrentHashMap<String,Long> getData(int count){
        return LongStream.rangeClosed(1, count).boxed()
                .collect(Collectors.toConcurrentMap(i -> UUID.randomUUID().toString(), Function.identity(),
                        (o1, o2) -> o1, ConcurrentHashMap::new));
    }


    @GetMapping("wrong")
    @Wrong
    @SneakyThrows
    public String wrong(){
        //900 个元素
        ConcurrentHashMap<String, Long> map = getData(ITEM_COUNT - 100);
        log.info("init size:{}",map.size());

        ForkJoinPool forkJoinPool = new ForkJoinPool(THREAD_COUNT);
        forkJoinPool.execute(()-> IntStream.rangeClosed(1,10).parallel().forEach(i->{
            // 查询还需要补充多少个元素
            int gap = ITEM_COUNT - map.size();
            log.info("gap size:{}",gap);
            map.putAll(getData(gap));
        }));


        // 等待所有任务完成
        forkJoinPool.shutdown();
        forkJoinPool.awaitTermination(1, TimeUnit.HOURS);

        // 最后有多少个?
        log.info("finish size:{}",map.size());
        return "OK";
    }

    @GetMapping("/right")
    @Right
    @SneakyThrows
    public String right() {
        ConcurrentHashMap<String, Long> data = getData(ITEM_COUNT - 100);
        log.info("init size:{}",data.size());

        ForkJoinPool forkJoinPool = new ForkJoinPool(THREAD_COUNT);
        forkJoinPool.execute(()->IntStream.rangeClosed(1,10).parallel().forEach(i->{
            synchronized (data){
                int gap = ITEM_COUNT - data.size();
                log.info("gap size:{}",gap);
                data.putAll(getData(gap));
            }
        }));

        forkJoinPool.shutdown();
        forkJoinPool.awaitTermination(1,TimeUnit.HOURS);

        log.info("finish size:{}",data.size());
        return "OK";
    }


    /**
     * 性能不高,不推荐
     * @return
     */
    @Wrong
    @SneakyThrows
    private Map<String,Long> normalUse() {
        ConcurrentHashMap<String, Long> map = new ConcurrentHashMap<>(ITEM_COUNT);
        ForkJoinPool forkJoinPool = new ForkJoinPool(THREAD_COUNT);
        forkJoinPool.execute(()->IntStream.rangeClosed(1,LOOP_COUNT).parallel().forEach(i ->{
            String key = "item"+ThreadLocalRandom.current().nextInt(ITEM_COUNT);
            synchronized (map){
                if (map.containsKey(key)){
                    map.put(key,map.get(key)+1);
                }else {
                    map.put(key,1L);
                }
            }
        }));
        forkJoinPool.shutdown();
        forkJoinPool.awaitTermination(1,TimeUnit.HOURS);
        return map;
    }

    @Right
    @SneakyThrows
    private Map<String,Long> goodUse() {
        ConcurrentHashMap<String, LongAdder> map = new ConcurrentHashMap<>(ITEM_COUNT);
        ForkJoinPool forkJoinPool = new ForkJoinPool(THREAD_COUNT);
        forkJoinPool.execute(()->IntStream.rangeClosed(1,LOOP_COUNT).parallel().forEach(i ->{
            String key = "item"+ThreadLocalRandom.current().nextInt(ITEM_COUNT);
            map.computeIfAbsent(key,k->new LongAdder()).increment();
        }));
        forkJoinPool.shutdown();
        forkJoinPool.awaitTermination(1,TimeUnit.HOURS);
        return map.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, e->e.getValue().longValue()));
    }


    /**
     * 测试 {@link #goodUse()} 与 {@link #normalUse()} 性能差异
     * @return
     */
    @GetMapping("good")
    public String good() {
        StopWatch stopWatch = new StopWatch();
        stopWatch.start("normalUse");
        Map<String, Long> stringLongMap = normalUse();
        stopWatch.stop();
        Assert.isTrue(stringLongMap.size() == ITEM_COUNT,"normalUse size error");
        Assert.isTrue(stringLongMap.values().stream()
                .mapToLong(Long::longValue)
                .reduce(0,Long::sum)== LOOP_COUNT
                ,"normalUse count error"
        );
        stopWatch.start("goodUse");
        Map<String, Long> goodUse = goodUse();
        stopWatch.stop();
        Assert.isTrue(goodUse.size()== ITEM_COUNT,"goodUse size error");
        Assert.isTrue(goodUse.values().stream()
                .mapToLong(Long::longValue)
                .reduce(0,Long::sum) == LOOP_COUNT
                ,"goodUse count error"
        );
        log.info(stopWatch.prettyPrint());
        return "OK";
    }


}
